package com.atbeijing.D03;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.ProcessingTimeSessionWindows;
import org.apache.flink.streaming.api.windowing.time.Time;

/**
 *会话窗口
 * 连续收集数据,当超过窗口的过期时间没有产生数据就关闭窗口
 */
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .socketTextStream("192.168.20.49",9999)
                .map(new MapFunction<String, Integer>() {
                    @Override
                    public Integer map(String value) throws Exception {
                        return Integer.valueOf(value);
                    }
                })
                //分到一个流
                .keyBy(r -> true)
                //机器时间会话窗口,5s内没产生数据就会关闭
                .window(ProcessingTimeSessionWindows.withGap(Time.seconds(5)))
                //来一条数据聚合一次,聚合后的结果放入累加器,当窗口闭合时向下游发放累加器
                .reduce(new ReduceFunction<Integer>() {
                    @Override
                    public Integer reduce(Integer value1, Integer value2) throws Exception {
                        return value1+value2;
                    }
                })
                .print();

        env.execute();
    }
}
